package com.idroc.cdfs.service.meta;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.SortedMap;
import java.util.TreeMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.cdfs.node.impl.StorageNode;
import com.idroc.cdfs.node.impl.StorageNode.BerkeleyCollection;
import com.idroc.rpc.RPCFactory;
import com.idroc.rpc.RPCProvider;
import com.sleepycat.collections.StoredMap;

/**
 * 
 * @author www.tianya.cn
 * @creator 詹都林 (zhandl@hainan.net) 
 * Aug 11, 2009 2:17:54 PM
 * @memo:
 */
public class LocalRegistryEndpoint implements RegistryEndpoint {
	
	private static final Logger log = LoggerFactory.getLogger(LocalRegistryEndpoint.class);
	
	public static final String ENDPOINT_CATALOG = "endpoint_catalog";	
	public static final String ROOT_ENDPOINT = "root_endpoint";
	public static final String REGISTERED_ENDPOINT = "registered_endpoint";
	
	private Integer epNo = null;
	private String host = null;
	private Integer port = null;
	private boolean isRootEndpoint = false;
	private boolean alive = true;
	
	private transient RegistryEndpoint root = null;
	private transient TreeMap<Integer, RegistryEndpoint> rep = new TreeMap<Integer, RegistryEndpoint>();
	
	private transient StorageNode sn = null;
	private transient BerkeleyCollection coll = null;
	private transient StoredMap store = null;
	
	private transient int hashCode = 0;
	
	public LocalRegistryEndpoint() {}
		
	/**
	 * 
	 */
	public LocalRegistryEndpoint(Integer epNo, String host, Integer port, boolean isRoot, StorageNode sn) {
		
		this.epNo = epNo;
		this.host = host;
		this.port = port;
		this.sn = sn;
		this.coll = sn.getBerkeleyCollection(ENDPOINT_CATALOG, String.class, HashMap.class);
		this.store = this.coll.open();
		if (isRoot) {
			this.isRootEndpoint = true;
			this.root = this;
			saveRootEndpoint(this);
		} else {
			RegistryEndpoint ep = loadRootEndpoint();
			if (null != ep) {
				try {
					this.root = new ProxyRegistryEndpoint(ep);
				} catch (Exception ex) {
					
				}
			} 
		}
		
		// 暴露rpc
		InetSocketAddress endpoint = new InetSocketAddress(host, port);
		RPCProvider provider = RPCFactory.bindEndpoint(endpoint); 
		provider.bind(RegistryEndpoint.class, this);
	}

	/*
	 * 获取本地端点的编号
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#getEndpointNo()
	 */
	public Integer getEndpointNo() {
		return epNo;
	}

	/*
	 * 获取主机IP
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#getEndpointHost()
	 */
	public String getEndpointHost() {
		return host;
	}

	/*
	 * 获取主机端口
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#getEndpointPort()
	 */
	public Integer getEndpointPort() {
		return port;
	}

	/*
	 * 获取已经注册地端点
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#getRegisteredEndpoint()
	 */
	public TreeMap<Integer, RegistryEndpoint> getRegisteredEndpoint() {
		return rep;
	}

	/*
	 * 获取根端点
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#getRootEndpoint()
	 */
	public RegistryEndpoint getRootEndpoint() {
		return root;
	}

	/*
	 * 注册端点
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#registerEndpoint(com.idroc.cdfs.service.meta.RegistryEndpoint)
	 */
	public synchronized boolean registerEndpoint(RegistryEndpoint ep) {
		
		if (isRootEndpoint) {
			if (null != ep) { 
				// 在其他端点中注册
				for (Integer no : rep.keySet()) {
					try {
						RegistryEndpoint re = rep.get(no);
						re.addEndpoint(this, ep);
					} catch (Exception ex) {
						ex.printStackTrace();
					}
				}
				// 放入已注册端点集合
				return addEndpoint(root, ep);
			}
			return false;
		} else {
			// 在根端点注册
			return root.registerEndpoint(ep);
		}
	}

	/*
	 * 取消注册
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#unregisterEndpoint(java.lang.Integer)
	 */
	public synchronized boolean unregisterEndpoint(Integer epNo) {
		
		if (isRootEndpoint) {
			RegistryEndpoint ep = rep.remove(epNo);
			if (null != ep) {
				// 在其他端点中将要取消的端点取消
				for (Integer no : rep.keySet()) {
					try {
						ep = rep.get(no);
						ep.dropEndpoint(this, epNo);
					} catch (Exception ex) {
						ex.printStackTrace();
					}
				}
			}
		} else {
			return root.unregisterEndpoint(epNo);
		}
		return false;
	}

	/*
	 * 增加一个端点, 当且仅当当前端点指向的跟端点与发出要求的跟端点相同
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#addEndpoint(com.idroc.cdfs.service.meta.RegistryEndpoint, com.idroc.cdfs.service.meta.RegistryEndpoint)
	 */
	public boolean addEndpoint(RegistryEndpoint fromRoot, RegistryEndpoint newArrive) {
		
		if (null != newArrive && null != fromRoot && root.equals(fromRoot)) {
			newArrive = (this.equals(newArrive)) ? newArrive : new ProxyRegistryEndpoint(newArrive);
			rep.put(newArrive.getEndpointNo(), newArrive);
			saveRegEndpoints(rep); // 保存端点集合
			return true;
		}
		return false;
	}

	/*
	 * 移除一个端点, 当且仅当当前端点指向的跟端点与发出要求的跟端点相同
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#dropEndpoint(com.idroc.cdfs.service.meta.RegistryEndpoint, java.lang.Integer)
	 */
	public boolean dropEndpoint(RegistryEndpoint fromRoot, Integer epNo) {
		
		if (null != epNo && null != fromRoot && root.equals(fromRoot)) {
			RegistryEndpoint ep = rep.remove(epNo);
			if (null != ep) {
				saveRegEndpoints(rep); // 保存端点集合
			}
			return true;
		}
		return false;
	}
	
	/**
	 * 设置保存信息使用的存储节点
	 * @param sn -- 存储节点
	 */
	public void setStorageNode(StorageNode sn) {
		this.sn = sn;
	}
	
	/**
	 * 当 编号相同, 主机IP相同, 端口相同, 认为两个端点相同
	 */
	@Override
	public boolean equals(Object obj) { 
		
		if (null == obj) {
			return false;
		}
		if (obj instanceof RegistryEndpoint) {
			RegistryEndpoint ep = (RegistryEndpoint)obj;
			Integer no = ep.getEndpointNo();
			String host = ep.getEndpointHost();
			Integer port = ep.getEndpointPort();
			return epNo.equals(no) && this.host.equals(host) && this.port.equals(port);
		} else {
			return false;
		}
	}

	@Override
	public int hashCode() { 
		
		if (hashCode == 0) {
			
			hashCode = 17;
			hashCode = 37 * hashCode + epNo.hashCode();
			hashCode = 37 * hashCode + host.hashCode();
			hashCode = 37 * hashCode + port.hashCode();
		}
		return hashCode;
	}

	/*
	 * 重建根端点
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#rebuildRootEndpoint()
	 */
	public RegistryEndpoint rebuildRootEndpoint(boolean nested) {
		
		if (isRootEndpoint && checkAlive()) {
			return this; // 如果当前端点是根
		} else if (nested) { // 嵌套查询可行节点并重建
			SortedMap<Integer, RegistryEndpoint> tails = rep.tailMap(Integer.valueOf(root.getEndpointNo().intValue() + 1));
			// 优先取根端点后面的端点然后才是前面的端点作为候选端点集合
			SortedMap<Integer, RegistryEndpoint> candicates = (tails.size() > 0) ? tails : rep.headMap(root.getEndpointNo()); 
			// 遍历端点找一个可以取代当前根结点
			for (Integer no : candicates.keySet()) {		
				try {
					RegistryEndpoint next = candicates.get(no);
					RegistryEndpoint candicate = next.rebuildRootEndpoint(false);
					if (null != candicate) {
						return candicate;
					} 
				} catch (Exception ex) {
					ex.printStackTrace();
				}
			}
			return null;
		} else { // 非嵌套查询重建
			SortedMap<Integer, RegistryEndpoint> tails = rep.tailMap(Integer.valueOf(root.getEndpointNo().intValue() + 1));
			// 优先取根端点后面的第一个端点然后才是前面的第一个端点作为候选端点
			RegistryEndpoint candicate = (tails.size() > 0) ? tails.get(tails.firstKey()) : rep.get(rep.firstKey());
			if (this.equals(candicate) && checkAlive()) {
				 
				// 通知端点更换根端点
				for (Integer no : rep.keySet()) {
					try { 
						RegistryEndpoint elem = rep.get(no);
						elem.replaceRootEndpoint(this); 
					} catch (Exception ex) {
						ex.printStackTrace();
					}
				}
				return this;
			}
			return null;
		}
	}
	
	/**
	 * 保存根端点到本地数据库
	 * @param root -- 根端点
	 * @return true, 保存成功
	 */
	private boolean saveRootEndpoint(RegistryEndpoint root) {
		
		if (null == root) {
			return false;
		}
		try {
			HashMap wrapper = new HashMap();
			wrapper.put(ROOT_ENDPOINT, root);
			store.put(ROOT_ENDPOINT, wrapper);
			return true;
		} catch (Exception ex) { /* swallow exception */ }
		return false;
	}
	
	/**
	 * 加载本地数据库中的根端点
	 * @return 根端点; null 如果在本地数据库不存在根端点
	 */
	private RegistryEndpoint loadRootEndpoint() {
		
		try {
			HashMap wrapper = (HashMap)store.get(ROOT_ENDPOINT);
			if (null == wrapper) {
				return null;
			}
			return (RegistryEndpoint)wrapper.get(ROOT_ENDPOINT); 
		} catch (Exception ex) {
			String errorMsg = "加载本地数据库中的根端点异常";
			throw new RuntimeException(errorMsg, ex);
		}
	}
	
	/**
	 * 保存已经注册地端点到本地数据库
	 * @param eps -- 已注册端点集合
	 * @return -- true, 保存成功
	 */
	private boolean saveRegEndpoints(TreeMap<Integer, RegistryEndpoint> eps) {
		
		if (null == eps || eps.size() == 0) {
			return false;
		}
		try {
			HashMap wrapper = new HashMap();
			wrapper.put(REGISTERED_ENDPOINT, eps);
			store.put(REGISTERED_ENDPOINT, wrapper);
			return true;
		} catch (Exception ex) { /* swallow exception */ }
		return false;
	}
	
	/**
	 * 加载在本地数据库的已注册端点集合
	 * @return -- 已注册端点集合; null, 如果本地数据库不存在此集合
	 */
	private TreeMap<Integer, RegistryEndpoint> loadRegEndpoints() {
		
		try {
			HashMap wrapper = (HashMap)store.get(REGISTERED_ENDPOINT);
			if (null == wrapper) {
				return null;
			}
			return (TreeMap<Integer, RegistryEndpoint>)wrapper.get(REGISTERED_ENDPOINT);
		} catch (Exception ex) {
			String errorMsg = "加载在本地数据库的已注册端点集合异常";
			throw new RuntimeException(errorMsg, ex);
		}
	}

	/*
	 * 
	 * @see com.idroc.cdfs.service.meta.RegistryEndpoint#checkAlive()
	 */
	public boolean checkAlive() { 
		return alive;
	}

	public boolean replaceRootEndpoint(RegistryEndpoint ep) {
		
		if (this.equals(ep)) {
			this.isRootEndpoint = true;
			this.root = this; 
		} else {
			ProxyRegistryEndpoint pep = new ProxyRegistryEndpoint(ep);
			this.root = pep; 
		}
		return saveRootEndpoint(root); 
	}
	
	@Override
	public String toString() {
		
		StringBuilder sb = new StringBuilder();
		sb.append("LocalRegistryEndpoint[");
		sb.append("epNo:").append(epNo).append(",");
		sb.append("host").append(host).append(",");
		sb.append("port:").append(port);
		sb.append("]");
		return sb.toString();
	}
}
